package com.margo.rabbit.direct;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;

/**
 * 	https://www.cnblogs.com/refuge/p/10356750.html
 * 		direct 消费者
 * 	默认交换器，不申明交换器
 * @author fangxingbang
 *
 */
public class MargoRabbitDirectConsumer {
	public static void main(String[] args) throws IOException, TimeoutException {
		directDefaultExchangeAck();
	}
	
	
	/**
	 * 	不确认，消息还可以被消费，不会从消息队列删除
	 * @throws IOException
	 * @throws TimeoutException
	 */
	public static void directDefaultExchange() throws IOException, TimeoutException {
		String queueName="margo-direct";
		ConnectionFactory connectionFactory=new ConnectionFactory();
		connectionFactory.setHost("127.0.0.1");
		Connection connection=connectionFactory.newConnection();
		Channel channel=connection.createChannel();
		//默认队列
		channel.queueDeclare(queueName, false, false, false, null);
		
		channel.basicConsume(queueName, new DefaultConsumer(channel) {
			 @Override
	         public void handleDelivery(String consumerTag,
	                                       Envelope envelope,
	                                       AMQP.BasicProperties properties,
	                                       byte[] body) throws IOException {
				 String message = new String(body, "UTF-8");
				 System.out.println("Received dead letter["
                        +envelope.getRoutingKey()
                        +"]"+message);
			 }
		});
	}
	
	/**
	 * 	自动确认，消息从消息队列删除
	 * @throws IOException
	 * @throws TimeoutException
	 */
	public static void directDefaultExchangeAutoAck() throws IOException, TimeoutException {
		String queueName="margo-direct";
		ConnectionFactory connectionFactory=new ConnectionFactory();
		connectionFactory.setHost("127.0.0.1");
		Connection connection=connectionFactory.newConnection();
		Channel channel=connection.createChannel();
		//申明队列
		channel.queueDeclare(queueName, false, false, false, null);
		
		channel.basicConsume(queueName, true, new DefaultConsumer(channel) {
			@Override
            public void handleDelivery(String consumerTag,
                                       Envelope envelope,
                                       AMQP.BasicProperties properties,
                                       byte[] body) throws IOException {
                String message = new String(body, "UTF-8");
                System.out.println("Received dead letter["
                        +envelope.getRoutingKey()
                        +"]"+message);
            }
		});
	}
	
	/**
	 * 	手动确认 
	 * @throws IOException
	 * @throws TimeoutException
	 */
	public static void directDefaultExchangeAck() throws IOException, TimeoutException {
		String queueName="margo-direct";
		ConnectionFactory connectionFactory=new ConnectionFactory();
		connectionFactory.setHost("127.0.0.1");
		Connection connection=connectionFactory.newConnection();
		//默认队列
		Channel channel=connection.createChannel();
		//将Qos预取值设置为1,这表示设置broker每次只推送队列里面的一条消息到消费者,只有在确认这条消息"成功消费"后,才会继续推送	
		channel.basicQos(0, 1, false);
		channel.queueDeclare(queueName, false, false, false, null);
		
		channel.basicConsume(queueName, false, new DefaultConsumer(channel) {
			 @SuppressWarnings("static-access")
			 @Override
			 public void handleDelivery(String consumerTag,
                                       Envelope envelope,
                                       AMQP.BasicProperties properties,
                                       byte[] body) throws IOException {
                String message = new String(body, "UTF-8");
                System.out.println("Received dead letter["
                        +envelope.getRoutingKey()
                        +"]"+message);
                /**
                 * deliveryTag 传递标签,ulong 类型.它的范围隶属于每个信道.因此必须在收到消息的相同信道上确认.
                 *	 不同的信道将导致“未知的传递标签”协议异常并关闭通道.
                 * multiple 确认一条消息还是多条.false 表示只确认 e.DelivertTag 这条消息,
                 * true表示确认 小于等于 e.DelivertTag 的所有消息 
                 */
                try {
                	//阻塞，测试 basicQos
					Thread.currentThread().sleep(1000);
				} catch (InterruptedException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
   			 	channel.basicAck(envelope.getDeliveryTag(), false);
            }

		});
	}
}
